package com.atliuzu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atliuzu.app.func.DimAsyncFunction;
import com.atliuzu.bean.TradeCourseBean;
import com.atliuzu.bean.TradeProvinceOrderBean;
import com.atliuzu.utils.ClickHouseUtil;
import com.atliuzu.utils.DateFormatUtil;
import com.atliuzu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author Administrator
 * @Time 2022-08-22-11:19
 */
public class DwsTradeProvinceOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1.1 状态后端设置
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                3, Time.days(1), Time.minutes(1)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage(
//                "hdfs://hadoop102:8020/ck"
//        );
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 2.从kafka dwd_trade_order_detail主题读取数据
        DataStreamSource<String> streamSource = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(
                "dwd_trade_order_detail", "group1"));

        //TODO 3.将数据流转化为json
        SingleOutputStreamOperator<JSONObject> jsonObjDS =
                streamSource.flatMap(new FlatMapFunction<String, JSONObject>() {
                    @Override
                    public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(value);
                        out.collect(jsonObject);
                    }
                });


        //TODO 4.将数据转为为javabean
        SingleOutputStreamOperator<TradeProvinceOrderBean> tradeProvinceDS =
                jsonObjDS.flatMap(new FlatMapFunction<JSONObject, TradeProvinceOrderBean>() {
            @Override
            public void flatMap(JSONObject value, Collector<TradeProvinceOrderBean> out) throws Exception {
                TradeProvinceOrderBean build = TradeProvinceOrderBean.builder()
                        .provinceId(value.getString("province_id"))
                        .userID(value.getString("user_id"))
                        .orderAmount(value.getDouble("final_amount"))
                        .orderCount(1L)
                        .ts(DateFormatUtil.toTs(value.getString("create_time"), true)).build();

                out.collect(build);
            }
        });

        //TODO 5.设水位线
        SingleOutputStreamOperator<TradeProvinceOrderBean> watermarksDS =
                tradeProvinceDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeProvinceOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderBean>() {
            @Override
            public long extractTimestamp(TradeProvinceOrderBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        //TODO 6.分组、开窗、聚合
        SingleOutputStreamOperator<TradeProvinceOrderBean> reduceDS =
                watermarksDS.keyBy(new KeySelector<TradeProvinceOrderBean, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(TradeProvinceOrderBean value) throws Exception {
                return new Tuple2<>(value.getUserID(), value.getProvinceId());
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10))).reduce(new ReduceFunction<TradeProvinceOrderBean>() {
            @Override
            public TradeProvinceOrderBean reduce(TradeProvinceOrderBean value1, TradeProvinceOrderBean value2) throws Exception {
                value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                return value1;
            }
        }, new WindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, Tuple2<String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple2<String, String> stringStringTuple2, TimeWindow window,
                              Iterable<TradeProvinceOrderBean> input, Collector<TradeProvinceOrderBean> out) throws Exception {
                //获取数据
                TradeProvinceOrderBean next = input.iterator().next();

                String sst = DateFormatUtil.toYmdHms(window.getStart());
                String edt = DateFormatUtil.toYmdHms(window.getEnd());

                //添加字段
                next.setEdt(edt);
                next.setStt(sst);
                next.setTs(System.currentTimeMillis());

                //写出数据
                out.collect(next);
            }
        });
        
        //TODO 7.关联维表添加字段
        //关联province
        SingleOutputStreamOperator<TradeProvinceOrderBean> provinceDS = AsyncDataStream.unorderedWait(reduceDS,
                new DimAsyncFunction<TradeProvinceOrderBean>("DIM_PROVINCE") {
            @Override
            public String getKey(TradeProvinceOrderBean input) {
                return input.getProvinceId();
            }

            @Override
            public void join(TradeProvinceOrderBean input, JSONObject dimInfo) {
                input.setProvinceName(dimInfo.getString("NAME"));
            }
        }, 100L, TimeUnit.SECONDS);

        //TODO 8.写入clickhouse
        provinceDS.print("===");
        provinceDS.addSink(ClickHouseUtil.getJdbcSink("insert into dws_trade_province_order_window values(?,?,?,?,?,?,?,?)"));

        //TODO 9.执行任务
        env.execute("dws_trade_province_order_window");

    }
}
